package com.cupdata.sms.kafka;

import com.cupdata.sms.config.ConfigProperties;
import com.cupdata.sms.core.CusKafkaProperties;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.Map;


/**
 * @author: zda
 * @description: TODO
 * @date: 2024/4/30 17:50
 */
@Component
public class KafkaConsumerService {
    @Autowired
    private ConfigProperties configProperties;

    public KafkaConsumer<String,byte[]> createConsumer(CusKafkaProperties cusKafkaProperties){
        Map<String, Object> properties = cusKafkaProperties.getOtherProps();
        if( properties.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)==null) properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,  false);
        if( properties.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG)==null) properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,  500 );
        if( properties.get(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG)==null) properties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,  60000 );
        if( properties.get(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG)==null) properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,  30000 );
        if( properties.get(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG)==null) properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,  10000 );
        if( cusKafkaProperties.getGroupId()==null){
            throw new RuntimeException("必须定义group_id");
        }
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,cusKafkaProperties.getGroupId());
        if( configProperties.getBoostrapServers()==null){
            throw new RuntimeException("必须定义bootstrapServers");
        }
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,configProperties.getBoostrapServers());
        return new KafkaConsumer(properties,
                new StringDeserializer(),new ByteArrayDeserializer());
    }
}
